package consumer

import (
	"context"
	"github.com/IBM/sarama"
	"go.uber.org/zap"
	"listenboos-common/kafka/model"
	"search/dao"
	"search/internal/config"
	"search/kafka/saramax"
)

type CollectConsumer struct {
	client         sarama.Client
	albumSearchDao *dao.AlbumElasticDAO
}

func NewCollectConsumer(c config.Config,
	client sarama.Client,
) *CollectConsumer {
	return &CollectConsumer{
		client:         client,
		albumSearchDao: dao.NewAlbumElasticDAO(),
	}
}
func (p *CollectConsumer) Start() error {
	cg, err := sarama.NewConsumerGroupFromClient("user_collect",
		p.client)
	if err != nil {
		zap.L().Error("KafkaAlbumAndTrackPlaySum系统kafka消息记录消费错误", zap.Error(err))
		return err
	}
	go func() {
		err = cg.Consume(context.Background(),
			[]string{"user_collect"},
			saramax.NewHandler[model.TrackModel](p.CollectConsume))
		if err != nil {
			// 记录日志
			zap.L().Error("KafkaAlbumAndTrackPlaySum系统kafka消息记录消费错误", zap.Error(err))
			//	return
		}
	}()
	return err
}

func (p *CollectConsumer) CollectConsume(msg *sarama.ConsumerMessage, t model.TrackModel) error {
	// 更新redis
	//ctx := context.Background()
	// 更新es
	////fmt.Println("插入es中")
	//err = p.albumSearchDao.UpdateAlbumCount(ctx, t.AlbumId, t.StatType, t.Count)
	//if err != nil {
	//	zap.L().Error("Consume update error:", zap.Error(err))
	//	return err
	//}

	// 需要更新一下消息总数

	return nil
}
func NewCollectConsumers(consumer *CollectConsumer) []Consumer {
	KafkaConsumers = append(KafkaConsumers, consumer)
	return KafkaConsumers
}
